package com.shujia.flink.core

import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

object Demo12IntervalJoin {
  def main(args: Array[String]): Unit = {
    /**
      * 动态拦截黑名单
      *
      */

    val env = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * id,date
      * 1,20200510
      * 2,20200511
      * 3,20200512
      * 4,20200513
      *
      */
    val ds = env.socketTextStream("node1", 6666)
      .map(line => {
        val split = line.split(",")
        (split(0), split(1), split(2).toLong)
      })
      .assignAscendingTimestamps(event => event._3) // 指定时间字段   毫秒级别
      .keyBy(_._1)

    /**
      * id,action
      * 1,0
      * 2,2
      * 3,1
      * 4,2
      *
      */
    val ds1 = env.socketTextStream("node1", 7777)
      .map(line => {
        val split = line.split(",")
        (split(0), split(1), split(2).toLong)
      })
      .assignAscendingTimestamps(event => event._3) // 指定时间字段   毫秒级别
      .keyBy(_._1)

    val joinDS = ds.intervalJoin(ds1)
      .between(Time.seconds(-5), Time.seconds(5))
      .process(new ProcessJoinFunction[(String, String, Long), (String, String, Long), (String, String, String)] {
        override def processElement(left: (String, String, Long), right: (String, String, Long), ctx: ProcessJoinFunction[(String, String,Long), (String, String,Long), (String, String, String)]#Context, out: Collector[(String, String, String)]): Unit = {
          val r = (left._1, left._2, right._2)
          out.collect(r)
        }
      })

    joinDS.print()


    env.execute()
  }
}


